/*
 * Copyright 2023 OPPO.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package com.oppo.cloud.syncer.consumer;

import com.alibaba.fastjson2.JSON;
import com.oppo.cloud.syncer.domain.DebeziumMessage;
import com.oppo.cloud.syncer.domain.DebeziumMessageWithSchema;
import com.oppo.cloud.syncer.util.DataUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Component;

import java.util.Map;
import java.util.StringJoiner;

/**
 * 消费者
 */
@Slf4j
@Component
public class SyncConsumer {

    @Autowired
    @Qualifier("diagnoseJdbcTemplate")
    private JdbcTemplate jdbcTemplate;

    /**
     * Receive and handle sql data
     */
    @KafkaListener(topics = "${spring.kafka.sync-topic}", containerFactory = "kafkaListenerContainerFactory")
    public void receive(@Payload String message,
                        @Header(KafkaHeaders.RECEIVED_PARTITION_ID) int partition,
                        @Header(KafkaHeaders.RECEIVED_TOPIC) String topic,
                        org.apache.kafka.clients.consumer.Consumer consumer) {

        log.debug(String.format("From partition %d: %s", partition, message));

        if (message == null || message.isEmpty() || message.equals("null")) {
            // 处理空消息的情况
            log.info("Received an empty message, ignoring it.");
            consumer.commitAsync();
            return;
        }

        DebeziumMessageWithSchema debeziumMessage = JSON.parseObject(message, DebeziumMessageWithSchema.class);
        String table = debeziumMessage.getPayload().getSource().get("table");
        Map<String, String> data;
        if (debeziumMessage.getPayload().getOp() == "d"){
            data = debeziumMessage.getPayload().getBefore();
        }else {
            data = debeziumMessage.getPayload().getAfter();
        }


        // 使用 data 拼接 sql 语句
        insertData(data, table);




        consumer.commitAsync();
    }
    public void insertData(Map<String, String> data,String table) {
        StringBuilder columns = new StringBuilder();
        StringBuilder values = new StringBuilder();
        Object[] params = new Object[data.size()];

        int i = 0;
        for (Map.Entry<String, String> entry : data.entrySet()) {
            String field = entry.getKey();
            String value = entry.getValue();
            columns.append(field).append(", ");
            values.append("?, ");
            if (field.equals("update_time") || field.equals("create_time")){
                params[i] = DataUtil.parseDate(DataUtil.parseLong(value));
            }else {
                params[i] = value;
            }
            i++;
        }

        // 去掉最后一个逗号和空格
        columns.setLength(columns.length() - 2);
        values.setLength(values.length() - 2);

        String sql = "INSERT INTO  public." + table + " (" + columns + ") VALUES (" + values + ")";
        log.info(sql);
        jdbcTemplate.update(sql, params);
    }
    private String genericSql(Map<String, String> data, String table) {
        StringBuilder sb = new StringBuilder();
        sb.append("INSERT INTO ").append(table).append(" (");
        StringJoiner keyJoiner = new StringJoiner("\",\""," ( "," ) ");
        StringJoiner valueJoiner = new StringJoiner("\",\""," ( "," ) ");
        for (Map.Entry<String, String> element : data.entrySet()) {
            keyJoiner.add(element.getKey());
            valueJoiner.add(element.getValue());
        }


        return null;
    }


}
